﻿using Confluent.Kafka;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Text;
using System.Threading;

namespace Surging.Core.EventBusKafka.Implementation
{
    public class KafkaConsumerPersistentConnection : KafkaPersistentConnectionBase
    {
        private readonly ILogger<KafkaConsumerPersistentConnection> _logger;
        private ConcurrentBag<IConsumer<string, string>> _consumerClients;
        private IConsumer<string, string> _consumerClient;
        //private readonly IDeserializer<string> _stringDeserializer;
        bool _disposed;

        public KafkaConsumerPersistentConnection(ILogger<KafkaConsumerPersistentConnection> logger)
            : base(logger, AppConfig.KafkaConsumerConfig)
        {
            _logger = logger;
            //_stringDeserializer = new StringDeserializer(Encoding.UTF8);
            _consumerClients = new ConcurrentBag<IConsumer<string, string>>();
        }

        public override bool IsConnected => _consumerClient != null && !_disposed;

        public override Action Connection(IEnumerable<KeyValuePair<string, object>> options)
        {
            return () =>
            {
                List<KeyValuePair<string, string>> option = new List<KeyValuePair<string, string>>();
                foreach (var item in options)
                    option.Add(new KeyValuePair<string, string>(item.Key, item.Value.ToString()));
                _consumerClient = new ConsumerBuilder<string, string>(option).SetErrorHandler(OnConsumeError).Build();//, null, _stringDeserializer);
                //_consumerClient.OnConsumeError += OnConsumeError;
                //_consumerClient.OnError += OnConnectionException;
                _consumerClients.Add(_consumerClient);

            };
        }

        public void Listening(TimeSpan timeout)
        {
            if (!IsConnected)
            {
                TryConnect();
            }
            CancellationTokenSource cts = new CancellationTokenSource();
            Console.CancelKeyPress += (sender, e) =>
            {
                //prevent the process from terminating.
                e.Cancel = true;
                cts.Cancel();
            };
            //是否消费成功
            bool isOK = false;
            //result
            ConsumeResult<string, string> consumeResult = null;
            while (true)
            {
                foreach (var client in _consumerClients)
                {
                    isOK = false;
                    try
                    {
                        //如果需要从指定位置消费，可以参考下面这段代码；下面这段代码别乱起用（慎重）
                        //consumer.Assign(new TopicPartitionOffset(queueName, 0, Offset.Beginning));
                        consumeResult = client.Consume(cts.Token);
                        if (consumeResult.IsPartitionEOF)
                        {
                            //WriteLog($"Reached end of topic {consumeResult.Topic}, partition {consumeResult.Partition}, offset {consumeResult.Offset}.");
                            continue;
                        }
                        //接收到的消息记录
                        //消息消费
                        ProcessEvent?.Invoke(consumeResult.Message.Key, consumeResult.Message.Value);
                        //消费成功
                        isOK = true;
                        //提交方法向Kafka集群发送一个“提交偏移量”请求，并同步等待响应。
                        //与消费者能够消费消息的速度相比，这是非常慢的。
                        //一个高性能的应用程序通常会相对不频繁地提交偏移量，并且在失败的情况下被设计来处理重复的消息
                        client.Commit(consumeResult);
                        //消费成功Log记录
                        //WriteLog($"Consumed message '{consumeResult.Message.Value}' at: '{consumeResult.TopicPartitionOffset}'.");
                    }
                    catch (ConsumeException e)
                    {
                        isOK = false;
                        _logger.LogError(e, $"Error occured: {e.Error.Reason}");
                    }
                    catch (Exception ex)
                    {
                        isOK = false;
                        _logger.LogError(ex, $"Error occured: {ex.StackTrace}");
                    }

                    //消费失败后置处理
                    if (!isOK && consumeResult != null)
                    {
                        //消费失败代码逻辑处理
                        TryConnect();
                    }

                    //var consumeResult = client.Consume(cts.Token);
                    //if (consumeResult.IsPartitionEOF)
                    //{
                    //    continue;
                    //}
                    //if (consumeResult.Offset % 5 == 0)
                    //{
                    //    client.Commit(consumeResult);
                    //}

                    //client.Poll(timeout);
                    //if (!client.Consume(out Message<string, string> msg, (int)timeout.TotalMilliseconds))
                    //{
                    //    continue;
                    //}
                    //if (msg.Offset % 5 == 0)
                    //{
                    //    var committedOffsets = client.CommitAsync(msg).Result;
                    //}
                }
            }
        }

        public override object CreateConnect()
        {
            TryConnect();
            return _consumerClient;
        }

        private void OnConsumeError(IConsumer<string, string> message, Error e)
        {
            //var message =  e.Deserialize<string, string>(null, _stringDeserializer);
            if (_disposed) return;

            _logger.LogWarning($"An error occurred during consume the message; Topic:'{message.Name}'," +
                $"Message:'{message.ToString()}', Reason:'{e.Reason}'.");

            TryConnect();
        }

        private void OnConnectionException(object sender, Error error)
        {
            if (_disposed) return;

            _logger.LogWarning($"A Kafka connection throw exception.info:{error} ,Trying to re-connect...");

            TryConnect();
        }

        public override void Dispose()
        {
            if (_disposed) return;

            _disposed = true;

            try
            {
                _consumerClient.Dispose();
            }
            catch (IOException ex)
            {
                _logger.LogCritical(ex.ToString());
            }
        }
    }
}
